package com.xiashitech.agent.grpc;

import com.xiashitech.agent.cache.data.CacheData;
import com.xiashitech.agent.cache.data.CacheSync;
import com.xiashitech.agent.config.StartConfig;
import com.xiashitech.agent.instrumentation.db.statement.sanitizer.impl.XSSqlSanitizer;
import com.xiashitech.agent.utils.JsonUtil;
import com.xiashitech.interfaces.apiservice.dto.user.UserInfoRequest;
import com.xiashitech.interfaces.apiservice.dto.user.UserInfoResponse;
import com.xiashitech.interfaces.grpc.RPCUserInfoRequest;
import com.xiashitech.interfaces.grpc.RPCUserInfoResponse;
import com.xiashitech.interfaces.grpc.RPCUserServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.stub.StreamObserver;

public class XsUserClient {
    private final static XsUserClient client = new XsUserClient();

    public static XsUserClient getClient() {
        return client;
    }

    public void retrieveUserInfoGRPC(UserInfoRequest userRequest) {
        System.out.println(Thread.currentThread().getId() + ", retrieveUserInfoGRPC enter, StartConfig.agentService: " + StartConfig.getParamValueFromKey(StartConfig.agentService));
        ManagedChannel channelRef = XsChannelManager.getManager().buildStubOfChannel("127.0.0.1", 9988);
        System.out.println(Thread.currentThread().getId() + ", retrieveUserInfoGRPC, now channel identityHashCode: "+ System.identityHashCode(channelRef));
        System.out.println(Thread.currentThread().getId() + ", retrieveUserInfoGRPC, now userRequest identityHashCode: "+ System.identityHashCode(userRequest));

        StreamObserver<RPCUserInfoResponse> responseStreamObserver = new StreamObserver<RPCUserInfoResponse>() {
            @Override
            public void onNext(RPCUserInfoResponse value) {
                // 采用这方式，每次调用，应该都会把外部的这个引用和当前 this 对象绑定到一起
                System.out.println( Thread.currentThread().getId() + ", " + System.identityHashCode(userRequest) + " : " + userRequest + " ===>的响应为===> " + value.getUserInfoResponseJson());
                // 将json 转成对象;
                UserInfoResponse userInfoResponse = JsonUtil.convertStringToObject(value.getUserInfoResponseJson(),UserInfoResponse.class);
                // 获取 requestKey
                CacheData cacheData = CacheSync.getInstance().getCacheData(userRequest);
                if(cacheData != null) {
                    cacheData.setReturnValue(userInfoResponse);
                } else {
                    // FAILED，服务端推送过来的数据，客户端没有相应的缓存
                    System.out.println("the cache ISNOT exist, should init");
                    CacheSync.getInstance().init(XSSqlSanitizer.class, "supplementUserTracingForJob", userRequest, userInfoResponse);
                    new Thread(()->{
                        retrieveUserInfoGRPC(userRequest);
                    }).start();
                }
            }

            @Override
            public void onError(Throwable t) {
                System.out.println(" supplementUserTracingForJob response error");
            }

            @Override
            public void onCompleted() {
                System.out.println(" supplementUserTracingForJob response onCompleted");
            }
        };
        RPCUserServiceGrpc.RPCUserServiceStub stub = RPCUserServiceGrpc.newStub(channelRef);
        StreamObserver<RPCUserInfoRequest> requestStreamObserver = stub.biStreamUserInfo(responseStreamObserver);

        // 将 authFactRequest 对象建立映射，用于将推送回来的数据修改缓存
        String authFactRequestKey = String.valueOf(System.identityHashCode(userRequest));

        System.out.println(Thread.currentThread().getId() + ", requestStreamObserver begin post request ");
        // 将 requestStreamObserver 缓存起来，下次发送请求
        try {
            System.out.println("authFactRequest: " + userRequest + " ====> " + JsonUtil.convertObjectToString(userRequest));
            RPCUserInfoRequest rpcUserInfoRequest = RPCUserInfoRequest.newBuilder()
                    .setUserInfoRequestJson(JsonUtil.convertObjectToString(userRequest))
                    .build();
            System.out.println("Build rpcUserInfoRequest finished: " + rpcUserInfoRequest);
            requestStreamObserver.onNext(rpcUserInfoRequest);
            requestStreamObserver.onCompleted();
        }catch (Throwable t) {
            t.printStackTrace();
        }
        System.out.println(Thread.currentThread().getId() + ", requestStreamObserver post finished ");
    }
}
